package org.jactao.cluster;

import org.jactao.common.*;
import org.jactao.rpc.LChannel;
import org.jactao.common.LClientConfig;
import org.jactao.rpc.LContainer;
import org.jactao.rpc.LInvoker;
import org.jactao.cluster.loadbalance.LSelector;
import org.jactao.registry.LRegistryInfo;
import org.jactao.rpc.LResponseHolder;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by fanzetao on 2018/4/10.
 */
public class LClusterInvoker implements LInvoker {

    private LSelector selector;

    private List<LChannel> channels;

    private LReferConfig referConfig;

    public LClusterInvoker(LReferConfig referConfig) {
        this.referConfig = referConfig;
        this.selector = LSelector.get(referConfig.getLoadBalance());
    }

    private void findChannel(String serviceName, String version) {

        LClientConfig clientConfig = LContainer.getInstance().getClientConfig();

        if(channels == null) {

            synchronized (this) {

                if(channels == null) {

                    channels = new ArrayList<>();

                    List<LRegistryInfo> registryInfoList = referConfig.getRegistry().get(serviceName, version);
                    if (registryInfoList == null || registryInfoList.isEmpty())
                        throw new LException(serviceName + ":" + version + " service no provider found");

                    for (LRegistryInfo registryInfo : registryInfoList) {
                        LChannel channel = LChannel.get(clientConfig.getProtocolName());
                        if (channel == null)
                            throw new LException(clientConfig.getProtocolName() + " protocol not support");

                        System.out.println("find " + serviceName + ":" + version + " service on " +
                                registryInfo.getAddress().getAddress().getHostAddress() + ":" + registryInfo.getAddress().getPort());

                        channel.setAddress(registryInfo.getAddress());
                        channel.setSerializer(clientConfig.getSerializer());
                        channel.setWeight(registryInfo.getWeight());
                        channels.add(channel);
                    }
                }
            }
        }
    }

    @Override
    public LResponse invoke(LRequest request) throws LException {
        LResponseFuture responseFuture = new LResponseFuture();
        LResponseHolder.get().put(request.getRequestId(), responseFuture);

        findChannel(request.getClazz().getName(), request.getVersion());

        LChannel channel = selector.select(channels);

        if(! channel.connected())
            channel.connect();

        channel.send(request);

        LResponse response;

        try {
            response = LResponseHolder.get().
                    get(request.getRequestId()).get(referConfig.getTimeout());

            response.setRequestId(request.getRequestId());
        }catch (LException e) {
            throw new LException(e);
        }catch (Exception e) {
            throw new LException(e);
        }

        if (response.getThrowable() == null) {
            return response;
        } else {
            throw new LException(response.getThrowable());
        }
    }
}
